package com.ncamc.pms.core.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.ncamc.pms.core.common.constants.ExceptionCodeEnum;
import com.ncamc.pms.core.common.constants.KettleConstants;
import com.ncamc.pms.core.entity.ParamDbConnection;
import com.ncamc.pms.core.entity.ParamKettleTransFile;
import com.ncamc.pms.core.exception.ApplicationException;
import com.ncamc.pms.core.service.KettleTransService;
import com.ncamc.pms.core.service.ParamDbConnectionService;
import com.ncamc.pms.core.service.ParamKettleTransFileService;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Service
public class KettleTransServiceImpl implements KettleTransService {

    /**
     * kettle文件所在目录
     * todo 可以改成系统参数
     */
    @Value("${kettle.dir}")
    private String kettleDir;
    @Autowired
    private ParamKettleTransFileService kettleTransFileService;
    @Autowired
    private ParamDbConnectionService dbConnectionService;

    @Override
    public void runKettleProcessor(String kettleCode) {
        // 获得ktr文件流
        ParamKettleTransFile kettleTransFile = kettleTransFileService.getKettleTransFile(kettleCode);
        if (kettleTransFile == null) {
            throw new ApplicationException(ExceptionCodeEnum.KETTLE_DOES_NOT_EXIST, kettleCode);
        }
        InputStream inputStream;
        try {
            inputStream = getKettleTransInputStream(kettleTransFile.getKettleUrl());
        } catch (Exception e) {
            throw new ApplicationException(ExceptionCodeEnum.READ_KETTLE_FILE_ERROR, e, kettleTransFile.getKettleUrl());
        }
        if (inputStream == null) {
            throw new ApplicationException(ExceptionCodeEnum.READ_KETTLE_FILE_ERROR, kettleTransFile.getKettleUrl());
        }
        // 查询所有的数据库配置
        Map<String, ParamDbConnection> dbParams = getDbConnMap(kettleTransFile.getSourceConnectCode(), kettleTransFile.getTargetConnectCode());
        if (MapUtil.isEmpty(dbParams) || dbParams.size() < 2) {
            throw new ApplicationException(ExceptionCodeEnum.DB_CONFIG_DOES_NOT_EXIST);
        }
        // 生成sql中的参数
        HashMap<String, String> sqlParams = getSqlParams(kettleTransFile, dbParams);
        try {
            // 组装ktr执行参数
            runKettleFile(inputStream, dbParams, sqlParams);
        } catch (ApplicationException e) {
            throw e;
        } catch (Exception e) {
            log.error("kettle执行失败: {}, kettleFile={}, sqlParams={}, dbParams={}", e.getMessage(), JSON.toJSONString(kettleTransFile), JSON.toJSONString(sqlParams), JSON.toJSONString(dbParams));
            throw new ApplicationException(ExceptionCodeEnum.KETTLE_EXEC_ERROR, e, kettleCode);
        }
        // 保存ktr执行结果和日期
        kettleTransFileService.updateLastTime(kettleCode, DateUtil.formatDate(new Date()));
    }

    private HashMap<String, String> getSqlParams(ParamKettleTransFile kettleTransFile, Map<String, ParamDbConnection> dbParams) {
        HashMap<String, String> sqlParams = new HashMap<>();
        // 最后一次执行成功时间
        if (kettleTransFile != null && StrUtil.isNotBlank(kettleTransFile.getLastTime())) {
            sqlParams.put(KettleConstants.KETTLE_LAST_TIME_KEY, kettleTransFile.getLastTime());
        }
        // 查询表名
        if (dbParams != null) {
            ParamDbConnection sourceDB = dbParams.get(KettleConstants.KETTLE_SOURCE_DB_KEY);
            if (sourceDB != null) {
                sqlParams.put(KettleConstants.TABLE_SPACE, sourceDB.getDbTbSpace());
            }
        }
        return sqlParams;
    }

    /**
     * 执行 kettle 文件
     */
    private void runKettleFile(InputStream inputStream, Map<String, ParamDbConnection> dbParams, HashMap<String, String> sqlParams) throws Exception {
        // 创建 TransMeta
        KettleEnvironment.init();
        TransMeta transMeta = new TransMeta(inputStream, null, true, null, null);
        // 设置数据库
        for (DatabaseMeta databaseMeta : transMeta.getDatabases()) {
            ParamDbConnection dbParam = dbParams.get(databaseMeta.getName());
            if (dbParam == null) {
                continue;
            }
            setDatabase(transMeta, databaseMeta, dbParam);
        }
        // 创建 Trans
        Trans trans = getTrans(transMeta);
        // 设置sql参数
        for (String key : sqlParams.keySet()) {
            String value = sqlParams.get(key);
            if (StrUtil.isNotBlank(value)) {
                trans.setParameterValue(key, value);
            }
        }
        // 执行
        trans.execute(null);
        trans.waitUntilFinished();
    }

    /**
     * 拼接 kettle 文件路径
     */
    private InputStream getKettleTransInputStream(String kettleUrl) throws IOException {
        File file = new File(kettleDir + '/' + kettleUrl);
        if (!file.isFile()) {
            return null;
        }
        return Files.newInputStream(file.toPath());
    }

    /**
     * 获得数据库参数
     *
     * @return key ktr文件中配置的数据库编码, value 数据库连接配置
     */
    private Map<String, ParamDbConnection> getDbConnMap(String sourceConnectCode, String targetConnectCode) {
        List<ParamDbConnection> list = dbConnectionService.getAllConn();
        if (CollUtil.isEmpty(list)) {
            return Collections.emptyMap();
        }
        Map<String, ParamDbConnection> dbMap = list.stream().filter(e -> StrUtil.isNotBlank(e.getDbConnectionCode())).collect(Collectors.toMap(e -> e.getDbConnectionCode(), e -> e, (a, b) -> a));
        HashMap<String, ParamDbConnection> map = new HashMap<>();
        ParamDbConnection sourceDb = dbMap.get(sourceConnectCode);
        if (sourceDb != null) {
            map.put(KettleConstants.KETTLE_SOURCE_DB_KEY, sourceDb);
        }
        ParamDbConnection targetDb = dbMap.get(targetConnectCode);
        if (targetDb != null) {
            map.put(KettleConstants.KETTLE_TARGET_DB_KEY, targetDb);
        }
        return map;
    }

    /**
     * 设置数据库连接
     */
    private void setDatabase(TransMeta transMeta, DatabaseMeta databaseMeta, ParamDbConnection dbParam) throws Exception {
        if ("GENERIC".equals(dbParam.getDbType())) {
            transMeta.setParameterValue(KettleConstants.KETTLE_PARAMS_IPANDPORTANDDBNAME, dbParam.getIp() + ":" + dbParam.getPort() + "/" + dbParam.getDbName());
            transMeta.setParameterValue(KettleConstants.KETTLE_PARAMS_SUSER, dbParam.getUsername());
            transMeta.setParameterValue(KettleConstants.KETTLE_PARAMS_SPWD, dbParam.getPassword());
            return;
        }
        databaseMeta.setName(dbParam.getDbConnectionCode());
        databaseMeta.setDatabaseType(dbParam.getDbType());
        databaseMeta.setHostname(dbParam.getIp());
        databaseMeta.setDBPort(dbParam.getPort());
        databaseMeta.setUsername(dbParam.getUsername());
        databaseMeta.setPassword(dbParam.getPassword());
        databaseMeta.setDBName(dbParam.getDbName());
        if ("GENERIC".equals(dbParam.getDbType())) {
            databaseMeta.setAccessType(0);
            databaseMeta.getAttributes().setProperty("CUSTOM_DRIVER_CLASS", "dm.jdbc.driver.DmDriver");
        }
        if ("ORACLE".equals(dbParam.getDbType())) {
            databaseMeta.setDBName("/" + dbParam.getDbName());
        }
        databaseMeta.addExtraOption(dbParam.getDbType(), "characterEncoding", "utf-8");
        databaseMeta.addExtraOption(dbParam.getDbType(), "useSSL", "false");
        // 用于生成批次ID
        databaseMeta.addExtraOption(dbParam.getDbType(), "AUTOINCREMENT_SQL_FOR_BATCH_ID", "UPDATE param_etl_log_trans_counter SET ID=LAST_INSERT_ID(ID+1)");
        databaseMeta.addOptions();
    }

    /**
     * 获取转换配置
     */
    private Trans getTrans(TransMeta transMeta) {
        transMeta.setGatheringMetrics(true);
        transMeta.setCapturingStepPerformanceSnapShots(true);
        Trans trans = new Trans(transMeta);
        trans.setMonitored(true);
        trans.setInitializing(true);
        trans.setPreparing(true);
        trans.setRunning(true);
        trans.setSafeModeEnabled(true);
        trans.setGatheringMetrics(true);
        return trans;
    }
}
